package com.xrui.hbase.tools;

import com.google.common.base.Preconditions;

import com.xrui.hbase.HBaseEntityId;
import com.xrui.hbase.filter.Filters;
import com.xrui.hbase.filter.RegexRowFilter;
import com.xrui.hbase.filter.RowFilter;
import com.xrui.hbase.filter.RowPrefixFilter;
import com.xrui.hbase.schema.RowKeyComponent;
import com.xrui.hbase.schema.TableLayout;
import org.apache.hadoop.util.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

public class ToolUtils {
    public static class RowFinder {
        private boolean mSpecified;
        private final byte[] mRowKeyBytes;
        private final RowFilter mRowKeyFilter;

        public RowFinder(byte[] rowKeyBytes, RowFilter rowKeyFilter, boolean specified) {
            mRowKeyBytes = rowKeyBytes;
            mRowKeyFilter = rowKeyFilter;
            mSpecified = specified;
        }

        public byte[] getRowKeyBytes() {
            return mRowKeyBytes;
        }

        public RowFilter getRowKeyFilter() {
            return mRowKeyFilter;
        }

        public boolean IsSpecified() {
            return mSpecified;
        }
    }
    public static RowFinder populateRowKey(
            final TableLayout tableLayout, final Map<String, String> components
    ) throws IOException {
        boolean seenNull = false;
        List<String> validComponentNames = new ArrayList<>();
        for (RowKeyComponent rowKeyComponent : tableLayout.getRowKeyFormat().getComponents()) {
            validComponentNames.add(rowKeyComponent.getName());
        }
        Map<String, String> convertibleComponents = new LinkedHashMap<>();
        Map<String, String> regexSearchComponents = new LinkedHashMap<>();
        for (Map.Entry<String, String> entry : components.entrySet()) {
            String name = Preconditions.checkNotNull(entry.getKey());
            Preconditions.checkArgument(
                validComponentNames.contains(name),
                String.format("parameter name %s is not a valid row key component", name)
            );
            String value = entry.getValue();
            boolean isNull = null == value || value.isEmpty();
            if (isNull) {
                seenNull = true;
            }

            if (!isNull) {
                if (!seenNull) {
                    convertibleComponents.put(name, value);
                } else {
                    regexSearchComponents.put(name, value);
                }
            }
        }

        // Row key prefix cannot calculated if not all hashed components provided
        long hashItemSize = tableLayout.getRowKeyFormat().getComponents().stream()
            .filter(RowKeyComponent::isHashed).count();
        if (convertibleComponents.size() < hashItemSize) {
            regexSearchComponents.putAll(convertibleComponents);
            convertibleComponents.clear();
        }

        RowFilter rowFilter = null;
        byte[] rowKeyBytes = null;
        boolean specified = false;
        if (!convertibleComponents.isEmpty()) {
            rowKeyBytes = HBaseEntityId.create(tableLayout, convertibleComponents.values().toArray()).getHBaseRowKey();
            if(convertibleComponents.size() < validComponentNames.size()) {
                rowFilter = new RowPrefixFilter(rowKeyBytes);
            } else {
                specified = true;
            }
        }

        if (!regexSearchComponents.isEmpty()) {
            List<String> anchors = new ArrayList<>();
            for (String componentName : validComponentNames) {
                String anchor;
                if (convertibleComponents.containsKey(componentName)) {
                    anchor = convertibleComponents.get(componentName);
                } else {
                    anchor = regexSearchComponents.getOrDefault(componentName, ".*");
                }
                anchors.add(anchor);
            }
            String delimiter = Pattern.quote(tableLayout.getRowKeyFormat().getDelimiter());
            String regex = String.format("^.*%s$", StringUtils.join(delimiter, anchors));
            if(rowFilter != null) {
                rowFilter = Filters.and(rowFilter, new RegexRowFilter(regex));
            } else {
                rowFilter = new RegexRowFilter(regex);
            }
        }

        return new RowFinder(rowKeyBytes, rowFilter, specified);
    }
}
